package com.atguigu.gmall.realtime.app.functions;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

/**
 * Author: tiancy
 * Date: 2021/12/3
 * Desc: 异步维度关联类
 * 模板方法设计模式：
 *      在父类中定义实现某一个功能的核心算法的骨架，具体的实现延迟到子类中去完成；
 *      这样做的好处是，在不改变父类核心算法骨架的前提下，每一个子类中都可以有自己不同的实现方式
 */

/*
    TODO 考虑到从dim维度层`异步的`查询具体的维度信息,需要将当前类写的通用一些. 不能将输入类型和输出类型直接写死.
      因此需要将我们当前的类中待处理的数据类型和返回值的数据类型 定义为`泛型`,再使用.
      - 泛型模板,需要先声明在使用,如果我们需要`定义泛型方法`,可以直接看`PhoenixUtil.queryList()`的使用方式,先声明再使用.
         public static <T> List<T> queryList(String sql, Class<T> clz) { .... }
      - 泛型类中的泛型模板,需要声明在类名后面,也就是这样 先声明泛型模板 DimAsyncFunction<T>,再指定要处理的数据类型和返回值的数据类型.
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T>{
    // 声明线程池对象.
    private ExecutorService executorService;

    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        executorService = ThreadPoolUtil.getInstance();
    }

    /**
     * 异步操作需要实现分发请求的具体方法,如何处理,怎么返回调用结果 ....
     * @param obj 流中处理对象
     * @param resultFuture
     * @throws Exception
     */
    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        /*
            开启多个线程,需要借助我们自定义的线程池对象`com.atguigu.gmall.realtime.utils.ThreadPoolUtil`,调用提交线程的方法.
            由于`创建线程池对象`这种是一个比较重的方法,因此我们只能声明一次,并在`open方法中创建`
            以前我们创建一个线程,会调用`start()`方法使线程执行,这里也是相同的道理.我们需要先从线程池中拿到一个具体的线程,再执行`submit()`来使线程执行
         */
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                /**--发送异步请求来完成维度关联的具体实现---*/
                try {
                    long start = System.currentTimeMillis();
                    /**
                        获取维度的主键,这里要特别注意,我们当前流中的数据类型并不确定,可能是订单,可能是物流相关的东西,因此这里不能直接写死,需要根据当前流中的对象取,当前对象是什么,就获取什么对象的主键Id.
                        如果说,我知道我当前有这么一个`具体功能的方法`,但是需要根据当前`传入`的数据类型,做不同的实现方式. ====> 直接搞一个抽象方法,我只是声明,不做具体实现,具体的实现过程使用的时候再指定即可.
                     */
                    String key = getKey(obj);
                    //根据维度的主键 查询维度对象
                    JSONObject dimInfoJsonObj = DimUtil.getDimInfo(tableName, key);
                    //将维度对象中的属性值  赋值给流中的业务对象相关属性
                    if (dimInfoJsonObj != null) {
                        /** 这里还要一个抽象方法,用来 `将通过phoenix中查询的维度表中的数据`赋值给`流中对象的属性`.由于不知道当前流中处理的对象是谁,因此只能先定义成抽象类. */
                        join(dimInfoJsonObj,obj);
                    }
                    long end = System.currentTimeMillis();
                    System.out.println("异步查询耗时：" +(end - start) + "毫秒");

                    //获取数据库交互的结果并发送给ResultFuture的回调函数
                    resultFuture.complete(Collections.singleton(obj));
                } catch (Exception e) {
                    e.printStackTrace();
                    throw new RuntimeException("异步的维度关联发生了异常");
                }
            }
        });
    }
}
